{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "406bc9d0",
   "metadata": {},
   "source": [
    "# Paho MQTT Python 客户端使用"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7839445c-a7ba-4548-aed5-8abb42b71420",
   "metadata": {},
   "source": [
    "## 1. 安装mqtt for python库\n",
    "```bash\n",
    "# 安装 paho-mqtt 2.X\n",
    "pip install paho-mqtt\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46073987",
   "metadata": {},
   "source": [
    "## 2. 一个简单的示例"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5358f78b-1ec7-4451-9f01-e58b9ce0fb68",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "mqtt client start loop......\n",
      "Connected with result code Success\n",
      "TEST/CLIENTS/DATA b'{\"adress\":\"123456\",\"name\":\"null\",\"type\":\"LCD_THERMOSTATE\",\"t_01\":\"0.00\",\"t_02\":\"0.00\",\"t_03\":\"0.00\",\"t_04\":\"0.00\",\"h_01\":\"0\",\"p_01\":\"0\",\"co2\":\"0\",\"ip\":\"192.168.0.100\",\"mask\":\"255.255.255.0\",\"mac\":\"F0-F5-BD-10-29-68\",\"state\":\"offline\"}'\n",
      "parse json:{'adress': '123456', 'name': 'null', 'type': 'LCD_THERMOSTATE', 't_01': '0.00', 't_02': '0.00', 't_03': '0.00', 't_04': '0.00', 'h_01': '0', 'p_01': '0', 'co2': '0', 'ip': '192.168.0.100', 'mask': '255.255.255.0', 'mac': 'F0-F5-BD-10-29-68', 'state': 'offline'}\n",
      "key = adress, value = 123456\n",
      "key = name, value = null\n",
      "key = type, value = LCD_THERMOSTATE\n",
      "key = t_01, value = 0.00\n",
      "key = t_02, value = 0.00\n",
      "key = t_03, value = 0.00\n",
      "key = t_04, value = 0.00\n",
      "key = h_01, value = 0\n",
      "key = p_01, value = 0\n",
      "key = co2, value = 0\n",
      "key = ip, value = 192.168.0.100\n",
      "key = mask, value = 255.255.255.0\n",
      "key = mac, value = F0-F5-BD-10-29-68\n",
      "key = state, value = offline\n"
     ]
    }
   ],
   "source": [
    "import paho.mqtt.client as mqtt\n",
    "import json\n",
    "\n",
    "# 连接服务器应答回调\n",
    "def on_connect(client, userdata, flags, reason_code, properties):\n",
    "    # 打印连接状态码\n",
    "    print(f\"Connected with result code {reason_code}\")\n",
    "    # 订阅主题 \"TEST/#\",这里订阅前缀为TEST/的所有主题\n",
    "    client.subscribe('TEST/#')\n",
    "\n",
    "# 收到服务器消息回调\n",
    "def on_message(client, userdata, msg):\n",
    "    # 打印订阅的主题消息\n",
    "    print(msg.topic + \" \" + str(msg.payload))\n",
    "    # 解析json\n",
    "    obj = json.loads(msg.payload)\n",
    "    print(f\"parse json:{obj}\")\n",
    "    # 遍历字典(dict)\n",
    "    for k, v in obj.items():\n",
    "        print(f\"key = {k}, value = {v}\")\n",
    "\n",
    "# 创建MQTT使用V2版本API\n",
    "mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)\n",
    "\n",
    "# 注册回调\n",
    "mqttc.on_connect = on_connect\n",
    "mqttc.on_message = on_message\n",
    "\n",
    "# 连接服务器，指定服务器地址和端口，这里我们使用公共服务器\n",
    "mqttc.connect('mqtt.eclipseprojects.io', 1883, 60)\n",
    "print('mqtt client start loop......')\n",
    "\n",
    "# 这里是堵塞循环\n",
    "mqttc.loop_forever()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eaca786a",
   "metadata": {},
   "source": [
    "## 3. 一个完整的例子\n",
    "#### 创建MQTT连接（TCP连接）\n",
    "#### 指定MQTT连接的服务器地址、端口号、主题，生成随机客户端ID。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "042c46fe",
   "metadata": {},
   "outputs": [],
   "source": [
    "from paho.mqtt import client as mqtt_client\n",
    "import random\n",
    "import time\n",
    "import logging\n",
    "import json\n",
    "\n",
    "broker = 'broker.emqx.io'\n",
    "port = 1883\n",
    "topic = \"python/mqtt\"\n",
    "client_id = f'python-mqtt-{random.randint(0, 1000)}'\n",
    "username = 'emqx'\n",
    "password = 'public'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c00b289",
   "metadata": {},
   "source": [
    "\n",
    "#### 创建连接到服务器的客户端对象，并注册连接回调函数 on_connect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "e658fa7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "def connect_mqtt():\n",
    "    def on_connect(client, userdata, flags, rc, properties):\n",
    "        if rc == 0:\n",
    "            print(\"connected to MQTT Broker!\")\n",
    "        else:\n",
    "            print(f\"failed to connect, return code {rc}\")\n",
    "    # 设置连接客户端ID\n",
    "    client = mqtt_client.Client(client_id = client_id, callback_api_version=mqtt_client.CallbackAPIVersion.VERSION2)\n",
    "    # 设置用户名和密码\n",
    "    # client.username_pw_set(username, password)\n",
    "    client.on_connect = on_connect\n",
    "    client.connect(broker, port, keepalive=120)\n",
    "    return client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "52775b14",
   "metadata": {},
   "source": [
    "#### 自动重连"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "355f1577",
   "metadata": {},
   "outputs": [],
   "source": [
    "FIRST_RECONNECT_DELAY = 1\n",
    "RECONNECT_RATE = 2\n",
    "MAX_RECONNECT_COUNT = 12\n",
    "MAX_RECONNECT_DELAY = 60\n",
    "\n",
    "FLAG_EXIT = False\n",
    "\n",
    "def on_disconnect(client, userdata, rc):\n",
    "    logging.info(f\"disconnected with result code: {rc}\")\n",
    "    reconnect_count, reconnect_delay = 0, FIRST_RECONNECT_DELAY\n",
    "    while reconnect_count < MAX_RECONNECT_COUNT:\n",
    "        logging.info(f\"reconnecting in {reconnect_count} seconds...\")\n",
    "        time.sleep(reconnect_delay)\n",
    "\n",
    "        try:\n",
    "            client.reconnect()\n",
    "            logging.info(\"reconnected successfully!\")\n",
    "            return\n",
    "        except Exception as err:\n",
    "            logging.error(f\"{err}. reconnect failed, retrying...\")\n",
    "\n",
    "        reconnect_delay *= RECONNECT_RATE\n",
    "        reconnect_delay = min(reconnect_delay, MAX_RECONNECT_DELAY)\n",
    "        reconnect_count += 1\n",
    "    logging.info(f\"reconnect failed after {reconnect_count} attempts.Exiting...\")\n",
    "    global FLAG_EXIT\n",
    "    FLAG_EXIT = True"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d93c23f",
   "metadata": {},
   "source": [
    "#### 发布消息"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "33568cb7",
   "metadata": {},
   "outputs": [],
   "source": [
    "def publish(client):\n",
    "    msg_count = 1\n",
    "    while not FLAG_EXIT:\n",
    "        msg_dict = {'msg':msg_count}\n",
    "        msg = json.dumps(msg_dict)\n",
    "        if not client.is_connected():\n",
    "            logging.error(\"publish:MQTT client is not connected!\")\n",
    "            time.sleep(1)\n",
    "            continue\n",
    "        result = client.publish(topic, msg)\n",
    "        status = result[0]\n",
    "        if status == 0:\n",
    "            print(f\"send '{msg}' to topic '{topic}'\")\n",
    "        else:\n",
    "            print(f\"failed to send message to topic {topic}\")\n",
    "        \n",
    "        msg_count += 1\n",
    "        if msg_count > 5:\n",
    "            break\n",
    "        \n",
    "        time.sleep(1)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4d95869",
   "metadata": {},
   "source": [
    "#### 订阅"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "a37a4b0e",
   "metadata": {},
   "outputs": [],
   "source": [
    "def subscribe(client: mqtt_client):\n",
    "    def on_message(client, userdata, msg):\n",
    "        print(f\"received '{msg.payload.decode()}' from '{msg.topic}' topic\")\n",
    "    \n",
    "    client.subscribe(topic)\n",
    "    client.on_message = on_message\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fd6d9b9",
   "metadata": {},
   "source": [
    "#### 测试代码"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "88b14217",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "connected to MQTT Broker!\n",
      "send '{\"msg\": 1}' to topic 'python/mqtt'\n",
      "received '{\"msg\": 1}' from 'python/mqtt' topic\n",
      "send '{\"msg\": 2}' to topic 'python/mqtt'\n",
      "received '{\"msg\": 2}' from 'python/mqtt' topic\n",
      "send '{\"msg\": 3}' to topic 'python/mqtt'\n",
      "received '{\"msg\": 3}' from 'python/mqtt' topic\n",
      "send '{\"msg\": 4}' to topic 'python/mqtt'\n",
      "received '{\"msg\": 4}' from 'python/mqtt' topic\n",
      "send '{\"msg\": 5}' to topic 'python/mqtt'\n"
     ]
    }
   ],
   "source": [
    "def run():\n",
    "    logging.basicConfig(format='%(asctimes - %(levelname)s:%(message)s)',level=logging.DEBUG)\n",
    "    client = connect_mqtt()\n",
    "    client.on_disconnect = on_disconnect\n",
    "    subscribe(client)\n",
    "    client.loop_start()\n",
    "    time.sleep(1)\n",
    "    if client.is_connected():\n",
    "        publish(client)\n",
    "    else:\n",
    "        client.loop_stop()\n",
    "\n",
    "    client.loop_stop()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    run()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
